package com.explorex.infra.puma.gateway.netty.ws.message.handler;


import com.explorex.infra.puma.gateway.netty.common.WaitingAckMessage;
import com.explorex.infra.puma.gateway.netty.common.ack.AckMessageManager;
import com.explorex.infra.puma.gateway.netty.common.enums.BusinessType;
import com.explorex.infra.puma.gateway.netty.common.enums.EnumBiz;
import com.explorex.infra.puma.gateway.netty.ws.WebSocketSession;
import com.explorex.infra.puma.gateway.netty.ws.enums.MessageTypeEnum;
import com.explorex.infra.puma.gateway.netty.ws.message.Message;
import com.explorex.infra.puma.gateway.netty.ws.message.MessageBody;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;

/**
 * @author: JacceYang
 * @createTime: 2022/07/19 2:29 下午
 * @description: 下行后消息ack 的结果处理器
 */
@Slf4j
public class MessageDownTypeProcessor extends AbstractBaseProcessor {

    private static final List<Integer> NEED_STATISTIC_ACK_BIZ_LIST = new ArrayList<>(5);

    static {
        NEED_STATISTIC_ACK_BIZ_LIST.add(EnumBiz.PUSH.getBizId());
        NEED_STATISTIC_ACK_BIZ_LIST.add(EnumBiz.IM.getBizId());
        NEED_STATISTIC_ACK_BIZ_LIST.add(EnumBiz.ROOM.getBizId());
        NEED_STATISTIC_ACK_BIZ_LIST.add(EnumBiz.IMPAAS.getBizId());
        NEED_STATISTIC_ACK_BIZ_LIST.add(EnumBiz.RENO.getBizId());
    }

    public static Message buildAck(Message messageFrame) {
        messageFrame.getHeader().setType(MessageTypeEnum.SILENCE.getType());
        messageFrame.getHeader().setTs(System.currentTimeMillis());
        return messageFrame;
    }

    @Override
    public short getMessageType() {
        return MessageTypeEnum.MESSAGE_DOWN_ACK.getType();
    }

    @Override
    public Message process(WebSocketSession session, Message messageFrame) {

        AckMessageManager.ackMessage(session, messageFrame.getHeader().getSeq(), BusinessType.valueOf(BusinessType.ROOM));
        //ack statistic,redcity集群不需要输出mq ack
        if (NEED_STATISTIC_ACK_BIZ_LIST.contains(messageFrame.getHeader().getBizId())) {
            processAck(session, messageFrame.getHeader().getSeq(), messageFrame.getHeader().getBizId());
        }
        return buildAck(messageFrame);
    }

    @Override
    public MessageBody parseBody(int contentStr, String msgBody) {
        return null;
    }

    @Override
    protected MessageBody doParseBody(int contentType, String msgBody) throws Exception {
        return null;
    }

    private void processAck(WebSocketSession session, long msgId, int bizId) {
        // info为null，说明Ack回复超过5min，上报ackTs=0, sendTs=0, 统计时会特殊处理
        try {
            WaitingAckMessage waitingAckMessage = AckMessageManager.ackMessage(session, msgId, BusinessType.valueOf((short) bizId));
        } catch (Exception e) {
            log.error("processAck  exception session={},msgId={},bizId={}", session, msgId, bizId);
        }
    }
}
